package cn.monkey.io.transport.netty;

import cn.monkey.io.transport.core.FluxReceive;
import cn.monkey.io.transport.netty.utils.ChannelHandlerUtils;
import io.netty.channel.Channel;
import io.netty.channel.EventLoop;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Operators;

import java.util.ArrayDeque;
import java.util.Queue;

public final class ChannelFluxReceive extends FluxReceive {
    private static final Logger log = LoggerFactory.getLogger(ChannelFluxReceive.class);
    final Channel channel;
    final EventLoop eventLoop;

    public ChannelFluxReceive(ChannelConnection connection) {
        this.channel = connection.channel();
        this.channel.pipeline().addLast(ChannelHandlerUtils.normalizeName(LastMessageChannelInboundHandler.class),
                new LastMessageChannelInboundHandler(this::onInboundNext));
        this.eventLoop = this.channel.eventLoop();
        this.disposed = false;
    }

    @Override
    protected void onInboundNext(Object msg) {
        if (this.disposed) {
            ReferenceCountUtil.release(msg);
            return;
        }
        if (this.receiver != null) {
            try {
                this.receiver.onNext(msg);
            } finally {
                ReferenceCountUtil.release(msg);
            }
            return;
        }
        Queue<Object> q = this.receiverQueue;
        if (q == null) {
            q = new ArrayDeque<>();
            this.receiverQueue = q;
        }
        if (q.offer(msg)) {
            drainReceiver();
            return;
        }
        if (log.isDebugEnabled()) {
            log.debug("FluxReceive.drainReceiver: queue is full, channelId: {}", this.channel.id());
        }
    }

    @Override
    public void request(long n) {
        if (Operators.validate(n)) {
            if (this.eventLoop.inEventLoop()) {
                this.receiverDemand = Operators.addCap(receiverDemand, n);
                this.drainReceiver();
            } else {
                this.eventLoop.execute(() -> {
                    this.receiverDemand = Operators.addCap(receiverDemand, n);
                    this.drainReceiver();
                });
            }
        }
    }


    @Override
    public void cancel() {
        if (this.eventLoop.inEventLoop()) {
            this.drainReceiver();
        } else {
            this.eventLoop.execute(this::drainReceiver);
        }
    }

    @Override
    public void subscribe(CoreSubscriber<? super Object> coreSubscriber) {
        if (this.eventLoop.inEventLoop()) {
            this.startReceiver(coreSubscriber);
        } else {
            this.eventLoop.execute(() -> this.startReceiver(coreSubscriber));
        }
    }
}
